<?php namespace App\Services;

use Carbon\Carbon;
use GuzzleHttp;
use Illuminate\Support\Facades\Log;
use Exception;
Use App\Events\Event;

class AliyunKafkaService
{
    protected $conf;
    protected $rk;
    protected $topic;

    //构造函数
    function __construct()
    {
        $conf = new \RdKafka\Conf();

        $conf->set('sasl.mechanisms', 'PLAIN');
        $conf->set('api.version.request', 'true');
        $conf->set('sasl.username', config('kafka.sasl_plain_username'));
        $conf->set('sasl.password', config('kafka.sasl_plain_password'));
        if (app()->environment('production') || app()->environment('integration')){
            $conf->set('security.protocol', 'SASL_SSL');
            $conf->set('ssl.ca.location', __DIR__ . '/../../config/ca-cert');
        }
        $conf->set('message.send.max.retries', 5);
        $conf->set('queue.buffering.max.ms', 1);
        $conf->set('socket.blocking.max.ms', 50);

        $this->rk = new \RdKafka\Producer($conf);
        $this->rk->setLogLevel(LOG_DEBUG);
        $this->rk->addBrokers(config('kafka.bootstrap_servers'));
        $this->topic = $this->rk->newTopic(config('kafka.topic_name'));
    }


    /**
     * @param Event $event
     */
    public function push($event)
    {
//        if (app()->environment('production') || app()->environment('integration')) {
        $body = [
            'tag' => $event->getTag(),
            'body' => json_encode($event)
        ];
        $bodyMessage = json_encode($body);
        $key = $this->gen_uuid();
        try {
            $a = $this->topic->produce(RD_KAFKA_PARTITION_UA, 0, $bodyMessage, $key);
            Log::info('[AliyunMQService->sendMessage] info: [messageKey] ' . $key . '[message] ' . $bodyMessage);

        } catch (\Exception $exception) {
            Log::error('[AliyunMQService->sendMessage] error: ' . $exception->getCode() . ' message: ' . $exception->getMessage()
                . 'original message: ' . $bodyMessage);
        }
        $this->rk->poll(5);


//        while ($rk->getOutQLen() > 0) {
//            $rk->poll(5);
//        }
    }

    function gen_uuid() {
        return sprintf( '%04x%04x-%04x-%04x-%04x-%04x%04x%04x',
            // 32 bits for "time_low"
            mt_rand( 0, 0xffff ), mt_rand( 0, 0xffff ),

            // 16 bits for "time_mid"
            mt_rand( 0, 0xffff ),

            // 16 bits for "time_hi_and_version",
            // four most significant bits holds version number 4
            mt_rand( 0, 0x0fff ) | 0x4000,

            // 16 bits, 8 bits for "clk_seq_hi_res",
            // 8 bits for "clk_seq_low",
            // two most significant bits holds zero and one for variant DCE1.1
            mt_rand( 0, 0x3fff ) | 0x8000,

            // 48 bits for "node"
            mt_rand( 0, 0xffff ), mt_rand( 0, 0xffff ), mt_rand( 0, 0xffff )
        );
    }
}

